package com.sunzm.spark.core.exercise

import org.apache.commons.lang3.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object SparkRDDExercise {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("RDD转换类算子示例")
      .setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)

    //方法一：
    //method1(sc)
    method2(sc)


    sc.stop
  }

  def method2(sc: SparkContext) = {
    val dataRDD: RDD[String] = sc.textFile("data/spark/rdd/{click.log,exposure.log}" )

    val reduceRDD: RDD[(String, (Int, Int))] = dataRDD.map(line => {
      if (StringUtils.contains(line, "/c")) {
        //点击日志
        val fields: Array[String] = StringUtils.splitByWholeSeparatorPreserveAllTokens(line, "adId=")

        //INFO 2016-07-25 00:29:53 requestURI:/c?app=0&p=1&did=18005472&industry=469&adId=31
        //取31

        val adId = fields(1)

        (adId, (1, 0))
      } else {
        //曝光日志
        val fields: Array[String] = StringUtils.splitByWholeSeparatorPreserveAllTokens(line, "adId=")

        val adId = fields(1)

        (adId, (0, 1))
      }
    }).reduceByKey((t1, t2) => {

      (t1._1 + t2._1, t1._2 + t2._2)
    })

    //打印
   /*reduceRDD.foreach{
      case (adId, (clickCount, exposureCount)) => {
       println(s"adId: ${adId}, 点击次数: ${clickCount}, 曝光次数: ${exposureCount}")
    }}*/

    //保存到文件
    reduceRDD.saveAsTextFile("data/spark/rdd/clickresult/")

    //保存到mysql
    reduceRDD.foreachPartition(p => {
      //获取数据量连接

      p.foreach {
        case (adId, (clickCount, exposureCount)) => {
         //写到数据库
        }
      }

      //关闭数据库连接
    })

  }

  def method1(sc: SparkContext) = {
    val clickRDD: RDD[String] = sc.textFile("data/spark/rdd/click.log")
    val exposureRDD: RDD[String] = sc.textFile("data/spark/rdd/exposure.log")

    //求点击次数
    val clickResultRDD: RDD[(String, Int)] = clickRDD.map(line => {
      val fields: Array[String] = StringUtils.splitByWholeSeparatorPreserveAllTokens(line, "adId=")

      //INFO 2016-07-25 00:29:53 requestURI:/c?app=0&p=1&did=18005472&industry=469&adId=31
      //取31

      val adId = fields(1)

      (adId, 1)
    }).reduceByKey(_ + _)

    //(31,2) (32, 1)
    clickResultRDD.foreach(line => println(line))

    //求曝光次数
    val exposureResultRDD: RDD[(String, Int)] = exposureRDD.map(line => {
      //全外连接
      val fields: Array[String] = StringUtils.splitByWholeSeparatorPreserveAllTokens(line, "adId=")

      //INFO 2016-07-25 00:29:53 requestURI:/c?app=0&p=1&did=18005472&industry=469&adId=31
      //取31

      val adId = fields(1)

      (adId, 1)
    }).reduceByKey(_ + _)

    //(31,2) (32, 1)
    //exposureResultRDD.foreach(line => println(line))
    println("------方法一：结果------")

    val fullOuterJoinRDD: RDD[(String, (Option[Int], Option[Int]))] = clickResultRDD.fullOuterJoin(exposureResultRDD)

    val resultRDD: RDD[(String, Int, Int)] = fullOuterJoinRDD.map(t => {
      val adId = t._1

      val option: (Option[Int], Option[Int]) = t._2

      //点击次数
      val clickCount: Int = option._1.getOrElse(0)
      //曝光次数
      val exposureCount: Int = option._2.getOrElse(0)

      (adId, clickCount, exposureCount)
    })

    resultRDD.foreach(line => println(line))
  }

}
